/*
 * <pre>
 * Copyright (c) The Loose Coupling Team.
 * Licensed under the Apache License, Version 2.0 (the "License"); 
 * you may not use this file except in compliance with the License. 
 * You may obtain a copy of the License at 
 *
 *      http://www.apache.org/licenses/LICENSE-2.0 
 *
 * Unless required by applicable law or agreed to in writing, software 
 * distributed under the License is distributed on an "AS IS" BASIS, 
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 * See the License for the specific language governing permissions and 
 * limitations under the License.
 *</pre>
 */
package org.lc.eic.core.pipesandfilters.impl;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;

import org.lc.eic.core.pipesandfilters.Filter;
import org.lc.eic.core.pipesandfilters.Pipe;
import org.lc.eic.core.pipesandfilters.exception.ProcessingException;
import org.lc.eic.core.support.logging.DefaultLogger;
import org.lc.eic.core.support.logging.Logger;

/**
 *  A filter implementation that thread offs and delegate the processing directly to next pipe;  
 *	@author <a href="mailto:p.lycke@googlemail.com">PaLy</a>
 *
 * 
 */
public class ThreadOffFilter<I> implements Filter<I, I> {
	private static final Logger logger = new DefaultLogger(ThreadOffFilter.class);

	private Pipe<I> nextPipe;

	private BlockingQueue<I> blockingQueue = new SynchronousQueue<I>();
	private int threadCount = 1;
	
	private boolean initialized = false;

	

	private void createThreads() {
		for (int i = 0; i < threadCount; i++) {		
			Thread thread = new Thread(getRunnable());
			thread.setDaemon(true);
			thread.start();
		}
	}

	private Runnable getRunnable() {
		return new Runnable() {

			public void run() {
				while (true) {
					try {
						I i = blockingQueue.take();

						nextPipe.process(i);

					} catch (InterruptedException e) {
						logger.debug("Thread interrupted", e);
					} catch (ProcessingException e) {
						logger.warn("ProcessingException caught", e);
					}
				}

			}

		};
	}


	/**
	 * @see org.lc.eic.core.pipesandfilters.Pipe#process(java.lang.Object)
	 */
	public void process(I i)  {
		
		if(!initialized){
			createThreads();
			initialized=true;
		}
		
		try {
			blockingQueue.put(i);
		} catch (InterruptedException e) {
			logger.debug("Thread interrupted", e);
		}

	}

	
	
	
	/**
	 * @see org.lc.eic.core.pipesandfilters.Filter#setOutboundPipe(org.lc.eic.core.pipesandfilters.Pipe)
	 */
	public void setOutboundPipe(Pipe<I> pipe) {
		this.nextPipe = pipe;
		initialized = false;
		

	}

	public void setThreadCount(int threadCount) {
		this.threadCount = threadCount;
		initialized = false;
	}

}
